package com.bw.wyq.base;

import com.bw.wyq.constant.Constant;
import com.bw.wyq.util.SQLUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public abstract class BaseSqlApp {
    public void start(Integer port,Integer parallelism ,String groupId){
        // 1. 创建环境

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        System.setProperty("HADOOP_USER_NAME","hadoop");
        // 创建表环境
        StreamTableEnvironment tableEnv  = StreamTableEnvironment.create(env);
        // 2. 设置并行度
        env.setParallelism(parallelism);
        // 3. 设置CK
        // 1.4.1 设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
        // 1.4.2 开启 checkpoint
//        env.enableCheckpointing(5000);
        // 1.4.3 设置 checkpoint 模式: 精准一次
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        // 1.4.4 checkpoint 存储
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall2023/stream/" + groupId);
//        // 1.4.5 checkpoint 并发数
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//        // 1.4.6 checkpoint 之间的最小间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
//        // 1.4.7 checkpoint  的超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(10000);
//        // 1.4.8 job 取消时 checkpoint 保留策略
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(RETAIN_ON_CANCELLATION);
        // 4、处理
        handle(env,tableEnv,groupId);

    }

    public abstract void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv,String groupId);

    // 读取Kafka topic_db数据
    public void readOdsDb(StreamTableEnvironment tableEnv,String groupId){
        // 用sql过滤出评论数据
        tableEnv.executeSql(SQLUtil.getKafkaTopicDb(Constant.TOPIC_DB));
    }

    //读取HBase的base_dic字典表
    public void createBaseDic(StreamTableEnvironment tableEnv){
        tableEnv.executeSql("CREATE TABLE base_dic (\n" +
                " rowkey STRING,\n" +
                " info ROW<dic_name STRING>,\n" +
                " PRIMARY KEY (rowkey) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'hbase-2.2',\n" +
                " 'table-name' = 'gmall:dim_base_dic',\n" +
                " 'zookeeper.quorum' = '"+ Constant.HBASE_ZOOKEEPER_QUORUM+"'\n" +
                ")");
    }
}
